package io.kiki.sba.registry.server.metadata.lease.dataStoreNode;

import io.kiki.sba.registry.common.model.metaserver.Lease;
import io.kiki.sba.registry.common.model.metaserver.cluster.VersionedList;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.DataStoreNode;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.lifecycle.impl.LifecycleHelper;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.cluster.node.NodeAdded;
import io.kiki.sba.registry.server.metadata.cluster.node.NodeRemoved;
import io.kiki.sba.registry.server.metadata.lease.impl.AbstractEvictableFilterableLeaseManager;
import io.kiki.sba.registry.server.metadata.monitor.Metrics;
import io.kiki.sba.registry.server.metadata.monitor.dataStoreNode.DataStoreNodeStats;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


@Component
public class DataStoreNodeManagerImpl extends AbstractEvictableFilterableLeaseManager<DataStoreNode> implements DataStoreNodeManager {

    private static final Logger logger = LoggerFactory.getLogger(DataStoreNodeManagerImpl.class);
    private final Map<String, DataStoreNodeStats> map = new ConcurrentHashMap<>();
    @Getter
    @Autowired
    private ServerConfig serverConfig;
    @Autowired
    private SlotManager slotManager;


    public DataStoreNodeManagerImpl() {
    }


    public DataStoreNodeManagerImpl(ServerConfig serverConfig, MetadataLeaderService metadataLeaderService) {
        this.serverConfig = serverConfig;
        this.metadataLeaderService = metadataLeaderService;
    }


    @PostConstruct
    public void postConstruct() throws Exception {
        LifecycleHelper.initializeIfPossible(this);
        LifecycleHelper.startIfPossible(this);
    }


    @PreDestroy
    public void preDestory() throws Exception {
        LifecycleHelper.stopIfPossible(this);
        LifecycleHelper.disposeIfPossible(this);
    }

    @Override
    public void register(Lease<DataStoreNode> lease) {
        super.register(lease);
        notifyObservers(new NodeAdded<>(lease.getRenewal()));
    }

    @Override
    public boolean cancel(Lease<DataStoreNode> lease) {
        boolean result = super.cancel(lease);
        if (result) {
            notifyObservers(new NodeRemoved<>(lease.getRenewal()));
            Metrics.Heartbeat.onDataEvict(lease.getRenewal().getIp());
        }
        return result;
    }

    @Override
    public boolean renew(DataStoreNode renewal, int leaseDuration) {
        Metrics.Heartbeat.onDataHeartbeat(renewal.getIp());
        return super.renew(renewal, leaseDuration);
    }

    @Override
    protected int getIntervalMilli() {
        return serverConfig.getExpireCheckIntervalMillis();
    }

    @Override
    protected int getEvictBetweenMilli() {
        return serverConfig.getExpireCheckIntervalMillis();
    }


    @Override
    public void onHeartbeat(HeartbeatRequest<DataStoreNode> heartbeatRequest) {
        String dataServer = heartbeatRequest.getNode().getIp();
        map.put(dataServer, new DataStoreNodeStats(dataServer, heartbeatRequest.getSlotsEpoch(), heartbeatRequest.getSlotStatuses()));
        learnFromData(heartbeatRequest);
    }

    protected void learnFromData(HeartbeatRequest<DataStoreNode> heartbeatRequest) {
        if (!amILeader()) {
            logger.info("data server heartbeat on follower.leader is:{}", metadataLeaderService.getLeader());
            return;
        }

        if (!metadataLeaderService.isWarmuped()) {
            logger.info("leader:{} is warming up.", metadataLeaderService.getLeader());
            return;
        }

        if (heartbeatRequest.getSlots() == null) {
            logger.info("data server:{} heartbeat Slots is null.", heartbeatRequest.getNode().getIp());
            return;
        }

        Slots slots = heartbeatRequest.getSlots();
        slotManager.refresh(slots);
    }

    @Override
    public List<DataStoreNodeStats> getDataStoreNodeStatsList() {
        return Collections.unmodifiableList(new LinkedList<>(map.values()));
    }

    @Override
    public VersionedList<DataStoreNode> getDataStoreNodeInfo() {
        VersionedList<Lease<DataStoreNode>> versionedList = VersionedLeaseList();
        List<DataStoreNode> list = new ArrayList<>();
        versionedList.getList().forEach(lease -> {
            list.add(lease.getRenewal());
        });
        return new VersionedList<>(list, versionedList.getEpoch());
    }

    @Override
    public long getEpoch() {
        return currentEpoch.get();
    }


}
